package com.bw.flinkstreaming.state1.job4;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.UUID;

public class MapStateWithCountAvg extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Double>> {

    //初始化状态
    private MapState<String,Long> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor<String,Long>(
                "avg",
                String.class,Long.class);
        mapState = getRuntimeContext().getMapState(mapStateDescriptor);
    }

    //业务代码
    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, Double>> out) throws Exception {
        mapState.put(UUID.randomUUID().toString(),element.f1);

        //取出value
        ArrayList<Long> values = Lists.newArrayList(mapState.values());

        if (values.size() ==3) {
            long count = 0;
            double sum = 0;
            for (Long e:values) {
                count ++;
                sum += e;
            }
            double avg = sum / count;
            out.collect(Tuple2.of(element.f0,avg));

            //清除state
            mapState.clear();
        }
    }
}
